package com.lkh.compute
import com.google.gson.Gson
import com.lkh.common.Config
import kafka.serializer.StringDecoder
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.elasticsearch.client.RestClient
import org.apache.http.HttpHost
import org.apache.spark.{SparkConf, SparkContext}
import org.elasticsearch.action.index.{IndexRequest, IndexResponse}
import org.elasticsearch.common.xcontent.XContentType

object ComputeWeiboDataToES {
  def main(args: Array[String]): Unit = {


    val conf = new SparkConf()
      .setAppName("app")
      .setMaster("local[4]")

    val sc = new SparkContext(conf)
    //创建spark streaming上下文对象
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    ssc.checkpoint("data/checkpoint")

    val params = Map(
      "zookeeper.connect" -> Config.getString("kafka.zookeeper.connect"),
      "group.id" -> "asdasasssdasf",
      "auto.offset.reset" -> "smallest",
      "zookeeper.connection.timeout.ms" -> "10000"
    )
    val topics = Map("wz" -> 4)

    //读取kafka数据   评价表数据
    val userDS: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](
      ssc, params, topics, StorageLevel.MEMORY_AND_DISK_SER
    )

    userDS.map(_._2)
      .filter(_.split("\\|").length == 8)
      .foreachRDD(rdd => {
        rdd.foreachPartition(iter => {

          // 创建es连接
          val httpHost = HttpHost.create(Config.getString("es.connect"))
          val builder = RestClient.builder(httpHost)
          val client = new RestHighLevelClient(builder)


          iter.foreach(line => {
            val split = line.split("\\|")
            val id = split(0)
            val sent_id = split(1)
            val user_id = split(2)
            val source = split(3)
            val reposts_count = split(4)
            val comments_count = split(5)
            val attitudes_count = split(6)
            val text = split(7)

            //构建一个索引，指定id
            val request = new IndexRequest("wz").id(id)

            //构建son格式的数据
            val wz = WZ(id, sent_id, user_id, source, reposts_count, comments_count, attitudes_count, text)
            val s = new Gson().toJson(wz)

            request.source(s, XContentType.JSON)

            //插入数据   indexResponse 可以从中获取状态
            val indexResponse = client.index(request, RequestOptions.DEFAULT)
          })

          client.close()
        })
      })


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

  case class WZ(id: String, sent_id: String, user_id: String, source: String, reposts_count: String, comments_count: String, attitudes_count: String, text: String)

}
